#!/usr/bin/env bash

include "objectv5.sh"
include "output.sh"
include "test.sh"

ObjectV5::DefineClass ParallelV2 "
    --cargs=ncpu,on_success,on_failure
    --string=fifopath
    --string=ncpu,4
    --array=tasks
    --array=running_pids
    --string=on_success
    --string=on_failure
"

ObjectV5::DefineClass ParallelV2Task "
    --cargs=index,name,outfile,errfile
    --string=index
    --string=name
    --string=outfile
    --string=errfile
    --array=command
    --string=pid
    --string=success
"

declare -A PidToTask

function ParallelV2::ParallelV2()
{
    [ $# == 0 ] && assert_fail "this"
    local this="$1"; shift
    ParallelV2::fifopath "$this" "$(mktemp --dry-run "/tmp/ParallelV2.XXXXXXXX")"
    return
}

function ParallelV2::AddCommand()
{
    [ $# == 0 ] && assert_fail "this"
    local this="$1"; shift
    local fifopath="$(ParallelV2::fifopath "$this")"
    local name="$1"; shift
    local command="$1"; shift
    if [ "$name" == "" ]; then
        assert_fail "CommandName不能是空的"
        return 1
    fi

    if [ "$command" == "" ]; then
        assert_fail "Command不能是空的"
        return 1
    fi

    ParallelV2Task::New task{"$(ParallelV2::tasks "$this" len)", "$name", "$fifopath", "$fifopath.err"}
    ParallelV2Task::command task push "$command"
    ParallelV2::tasks "$this" push "$task"
    return
}

function ParallelV2::Run()
{
    [ $# == 0 ] && assert_fail "this"
    local this="$1"
    local fifopath="$(ParallelV2::fifopath "$this")"
    local ncpu="$(ParallelV2::ncpu "$this")"
    local on_success="$(ParallelV2::on_success "$this")"
    local on_failure="$(ParallelV2::on_failure "$this")"

    function ParallelV2::RunTask()
    {
        local task="$1"; shift
        local name="$(ParallelV2Task::name task)"
        local command="$(ParallelV2Task::command task all)"
        local outfile="$(ParallelV2Task::outfile task).${BASHPID}"
        local errfile="$(ParallelV2Task::errfile task).${BASHPID}"
        debuglog "task: $name"
        if ! (eval $command) 1>"${outfile}" 2>&1; then
            mv "${outfile}" "${errfile}"
        fi
        echo "$BASHPID" > "$fifopath"
        debuglog "task: $name done"
        exit
    }

    function ParallelV2::WaitingTask()
    {
        [ $# == 0 ] && assert_fail "this"
        local this="$1"; shift
        local pid
        while read pid; do
            debuglog "readed $pid"
            ParallelV2::running_pids "$this" eraseval "$pid"
            local atask="${PidToTask["$pid"]}"
            ParallelV2Task::outfile atask "$(ParallelV2Task::outfile atask).$pid"
            ParallelV2Task::errfile atask "$(ParallelV2Task::errfile atask).$pid"
            local errfile="$(ParallelV2Task::errfile atask)"
            if [ -e "$errfile" ]; then
                ParallelV2Task::success atask false
                if declare -f "$on_failure" 1>/dev/null; then
                    $on_failure $atask
                fi
            else
                ParallelV2Task::success atask true
                if declare -f "$on_success" 1>/dev/null; then
                    $on_success $atask
                fi
            fi
        done <<< $(cat < "$fifopath")
    }

    if ! mkfifo -m 777 "$fifopath"; then
        loge "mkfifo $fifopath失败"
        return 1
    fi

    local i
    for (( i = 0; i < $(ParallelV2::tasks "$this" len); )); do
        local task="$(ParallelV2::tasks "$this" "$i")"
        local name="$(ParallelV2Task::name task)"
        debuglog "process $i: ${name}"
        if (( $(ParallelV2::running_pids "$this" len) < $ncpu )); then
            ParallelV2::RunTask "$task" &
            ParallelV2Task::pid task "$!"
            PidToTask["$!"]="$task"
            ParallelV2::running_pids "$this" push "$!"
            debuglog "running $!"
            (( i++ ))
            continue
        fi

        debuglog "waiting $i: $name"
        ParallelV2::WaitingTask "$this"
    done

    while (( "$(ParallelV2::running_pids "$this" len)" > 0 )); do
        debuglog "waiting running_pids $(ParallelV2::running_pids "$this" len)"
        ParallelV2::WaitingTask "$this"
    done

    rm "$fifopath"
    return
}

function ParallelV2::GetSuccessTasks()
{
    [ $# == 0 ] && assert_fail "this"
    local this="$1"; shift
    for task in $(ParallelV2::tasks "$this" all); do
        if $(ParallelV2Task::success task); then
            echo "$task"
        fi
    done
}

function ParallelV2::GetFailureTasks()
{
    [ $# == 0 ] && assert_fail "this"
    local this="$1"; shift
    for task in $(ParallelV2::tasks "$this" all); do
        if ! $(ParallelV2Task::success task); then
            echo "$task"
        fi
    done
}

function ParallelV2::~ParallelV2()
{
    [ $# == 0 ] && assert_fail "this"
    local this="$1"; shift
    local task
    for task in $(ParallelV2::tasks "$this" all); do
        ParallelV2Task::Del task
    done
}

function ParallelV2Task::~ParallelV2Task()
{
    [ $# == 0 ] && assert_fail "this"
    local this="$1"; shift
    local outfile="$(ParallelV2Task::outfile "$this")"
    local errfile="$(ParallelV2Task::errfile "$this")"
    rm -rf "$outfile" "$errfile"
}

function ParallelV2::Test1()
{
    function ParallelV2::Test1::task()
    {
        local value="$(printf "%d.%03d" "$(( $1 / 4 ))" "$(( $1 * 1000 / 4 % 1000 ))")"
        debuglog "$1: sleep $value"
        sleep "$value"
        return
    }

    local stime="$(date +%s%N)"
    ParallelV2 parallel{}
    local i
    for (( i = 0; i < 16; i++ )); do
        ParallelV2::AddCommand parallel "$i" "ParallelV2::Test1::task $((16 - i))"
    done

    ParallelV2::Run parallel
    test::run_cmd echo $(( ($(date +%s%N) - $stime) / 1000 / 1000 ))
    test::should_in 8500 8800
    return
}
test::register_test ParallelV2::Test1

function ParallelV2::Test2()
{
    function ParallelV2::Test2::task()
    {
        local value="$(printf "%d.%03d" "$(( $1 / 4 ))" "$(( $1 * 1000 / 4 % 1000 ))")"
        debuglog "$1: sleep $value"
        sleep "$value"
        return
    }

    local stime="$(date +%s%N)"
    ParallelV2 parallel{}
    local i
    for (( i = 1; i <= 16; i++ )); do
        ParallelV2::AddCommand parallel "$i" "ParallelV2::Test2::task $i"
    done

    ParallelV2::Run parallel
    test::run_cmd echo $(( ($(date +%s%N) - $stime) / 1000 / 1000 ))
    test::should_in 10000 10300
    return
}
test::register_test ParallelV2::Test2

function ParallelV2::Test3()
{
    function ParallelV2::Test3::TaskMain()
    {
        sleep 0.1
        echo $1
        return $(($1 % 2))
    }

    local success_messages=""
    local failure_messages=""
    function ParallelV2::Test3::OnSuccess()
    {
        local task="$1"; shift
        success_messages+="$(cat $(ParallelV2Task::outfile task))"
    }

    function ParallelV2::Test3::OnFailure()
    {
        local task="$1"; shift
        failure_messages+="$(cat $(ParallelV2Task::errfile task))"
    }

    ParallelV2 parallel{2, ParallelV2::Test3::OnSuccess, ParallelV2::Test3::OnFailure}
    local i
    for (( i = 1; i <= 8; i++ )); do
        ParallelV2::AddCommand parallel "$i" "ParallelV2::Test3::TaskMain $i"
    done
    ParallelV2::Run parallel

    test::var_should_be success_messages "2468"
    test::var_should_be failure_messages "1357"
    return
}
test::register_test ParallelV2::Test3

function ParallelV2::Test4()
{
    function ParallelV2::Test4::TaskMain()
    {
        echo $1
        return $(($1 % 2))
    }

    ParallelV2 parallel{2}
    local i
    for (( i = 1; i <= 8; i++ )); do
        ParallelV2::AddCommand parallel "$i" "ParallelV2::Test4::TaskMain $i"
    done

    ParallelV2::Run parallel
    for task in $(ParallelV2::GetSuccessTasks parallel); do
        test::should_return_true  [ "$(cat $(ParallelV2Task::outfile task))" == "$(ParallelV2Task::name task)" ]
        test::should_return_true  $(ParallelV2Task::success task)
    done

    for task in $(ParallelV2::GetFailureTasks parallel); do
        test::should_return_true  [ "$(cat $(ParallelV2Task::errfile task))" == "$(ParallelV2Task::name task)" ]
        test::should_return_false $(ParallelV2Task::success task)
    done
    return
}
test::register_test ParallelV2::Test4
